package com.inspur.edp.ai.flow.core.engine;

import com.inspur.edp.ai.flow.api.ServiceContainer;
import com.inspur.edp.ai.flow.engine.*;
import com.inspur.edp.ai.flow.engine.strategy.*;
import com.inspur.edp.ai.flow.exception.FlowException;
import com.inspur.edp.ai.flow.metadata.AbsNode;
import com.inspur.edp.ai.flow.metadata.Edge;
import com.inspur.edp.ai.flow.spi.NodeExecuteResult;
import com.inspur.edp.ai.flow.spi.NodeExecutor;
import com.inspur.edp.ai.flow.spi.NodeExecutorFactory;
import io.iec.edp.caf.commons.utils.SpringBeanUtils;
import io.iec.edp.caf.commons.utils.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.function.Predicate;

/**
 * @author lizhaorui
 * @date 2025/9/4
 * @description
 */

public class FlowExecuteEngine {

    public void executeSubFlow(EngineContext context, Flow subFlow) {
        context.addFlow(subFlow);
        context.getFlowContext().setFlow(subFlow);
        AbsNode headNode = subFlow.getHeadNode();
        Map<String, Object> inputParamValues = context.getNodeInputParamValues(headNode);
        executeNode(context, headNode, inputParamValues);
    }

    public void executeStartNode(EngineContext context, AbsNode startNode, List<Object> args) {
        Map<String, Object> startNodeInputMap = new HashMap<>();
        startNodeInputMap.put("start", args);
        this.executeNode(context, startNode, startNodeInputMap);
    }

    public void executeNode(EngineContext context, String nodeId) {
        AbsNode node = context.getFlowContext().getFlow().getNodeById(nodeId);
        Map<String, Object> inputParamValues = context.getNodeInputParamValues(node);
        executeNode(context, node, inputParamValues);
    }

    public void executeMultiTurnNode(EngineContext context, String nodeId, List<Object> additionalObjects) {
        AbsNode node = context.getFlowContext().getFlow().getNodeById(nodeId);
        Map<String, Object> inputParamValues = context.getNodeInputParamValues(node);
        inputParamValues.put(EngineConstant.MULTI_TURN_ADDITIONAL_KEY_LABEL, additionalObjects);
        executeNode(context, node, inputParamValues);
    }

    public void executeNode(EngineContext context, AbsNode node, Map<String, Object> inputParamValues) {
        if (context.isReturned()) {
            return;
        }
        DataBus dataBus = context.getDataBus();
        NodeExecutor nodeExecutor = NodeExecutorFactory.get(node.getKind());
        context.getFlowContext().setDataBus(dataBus);
        NodeExecuteResult executeResult = nodeExecutor.execute(context.getFlowContext(), node, inputParamValues);
        if (executeResult.getVariablesToAdd() != null) {
            dataBus.put(node.getCode(), executeResult.getVariablesToAdd());
        }
        if (node.isReturnNode()) {
            context.setReturnValues(executeResult.getVariablesToAdd());
            context.setReturned(true);
        }
        if (executeResult.getVariablesToRemove() != null) {
            Map<String, Object> nodeOutMap = (Map<String, Object>) dataBus.get(node.getCode());
            if (nodeOutMap != null) {
                for (String variableCode : executeResult.getVariablesToRemove()) {
                    nodeOutMap.remove(variableCode);
                }
            }
        }
        FlowStrategy strategy = executeResult.getFlowStrategy();
        this.processFlowStrategy(context, strategy);
    }

    private void executeEdge(EngineContext context, Edge edge) {
        Flow flow = context.getFlowContext().getFlow();
        List<Edge> allInputEdges = flow.getInputEdges(edge.getTargetNodeId());
        //判断 当前节点 是否为汇聚节点
        if (allInputEdges.size() > 1) {
            this.addArriveEdge(context, edge, allInputEdges, false);
        } else {
            this.executeNode(context, edge.getTargetNodeId());
        }
    }

    private void processFlowStrategy(EngineContext context, FlowStrategy flowStrategy) {
        String strategyKind = flowStrategy.getKind();
        Flow flow = context.getFlowContext().getFlow();
        switch (strategyKind) {
            case DefaultFlowStrategy.KIND:
                DefaultFlowStrategy defaultFlowStrategy = (DefaultFlowStrategy) flowStrategy;
                List<Edge> outputEdges = flow.getOutputEdges(defaultFlowStrategy.getNodeId());
                if (outputEdges == null || outputEdges.size() == 0) {
                    FlowFinRecord endArrivedRecord = context.addFinNode(flow.getId(), defaultFlowStrategy.getNodeId());
                    if (endArrivedRecord != null) {
                        if (endArrivedRecord.getFinishedEndNodeIds().size() == endArrivedRecord.getEndNodeIds().size()) {
                            //流程切换
                            String parentFlowId = flow.getParentId();
                            if (!StringUtils.isEmpty(parentFlowId)) {
                                context.switchToParentFlow(flow);
                                //返回到循环节点
                                this.executeNode(context, flow.getId());
                            }
                            return;
                        }
                    }
                }
                this.executePort(context, defaultFlowStrategy.getNodeId(), defaultFlowStrategy.getPortId());
                return;
            case BranchFlowStrategy.KIND:
                BranchFlowStrategy branchFlowStrategy = (BranchFlowStrategy) flowStrategy;
                for (String ignorePortId : branchFlowStrategy.getIgnorePortIds()) {
                    this.emptyExecutePort(context, branchFlowStrategy.getNodeId(), ignorePortId);
                }
                this.executePort(context, branchFlowStrategy.getNodeId(), branchFlowStrategy.getBranchPortId());
                return;
            case ParallelFlowStrategy.KIND:
                //调用 task 执行器
                return;
            case SuspendFlowStrategy.KIND:
                SuspendFlowStrategy suspendFlowStrategy = (SuspendFlowStrategy) flowStrategy;
                String instanceId = context.getFlowContext().getFlowInstanceId();
                List<String> suspendNodes = (List<String>) context.getDataBus().get(EngineConstant.SUSPEND_NODES_KEY_LABEL);
                if (suspendNodes == null) {
                    suspendNodes = new ArrayList<>();
                }
                suspendNodes.add(suspendFlowStrategy.getNodeId());
                context.getDataBus().put(EngineConstant.SUSPEND_NODES_KEY_LABEL, suspendNodes);
                ServiceContainer.getService(StatefulSupporter.class).backup(instanceId, context.getDataBus());
                return;
            case LoopBodyFlowStrategy.KIND:
                LoopBodyFlowStrategy startSubFlowStrategy = (LoopBodyFlowStrategy) flowStrategy;
                Flow subFlow = new Flow();
                subFlow.setId(startSubFlowStrategy.getNodeId());
                subFlow.setNodes(startSubFlowStrategy.getNodes());
                subFlow.setEdges(startSubFlowStrategy.getEdges());
                subFlow.setParentId(flow.getId());
                this.executeSubFlow(context, subFlow);
                return;
            default:
                throw new FlowException("not support flow strategy kind " + flowStrategy.getKind());
        }
    }

    private void executePort(EngineContext context, String sourceNodeId, String portId) {
        Flow flow = context.getFlowContext().getFlow();
        List<Edge> outputEdges = flow.getOutputEdges(sourceNodeId, portId);
        if (outputEdges.size() == 0) {
            return;
        }
        if (outputEdges.size() == 1) {
            this.executeEdge(context, outputEdges.get(0));
        } else {
            List<Future> futures = new ArrayList<>();
            for (Edge edge : outputEdges) {
                Runnable runnable = () -> {
                    this.executeEdge(context, edge);
                };
                Future future = ThreadPoolService.getExecutor().submit(runnable);
                futures.add(future);
            }
            for (Future future : futures) {
                try {
                    future.get();
                } catch (Throwable throwable) {
                    throw new FlowException(throwable);
                }
            }
        }
    }

    private void emptyExecutePort(EngineContext context,
                                  String sourceNodeId,
                                  String portId) {
        Flow flow = context.getFlowContext().getFlow();
        List<Edge> outputEdges = flow.getOutputEdges(sourceNodeId, portId);
        for (Edge edge : outputEdges) {
            this.emptyExecuteEdge(context, edge);
        }
    }

    private void emptyExecuteNode(EngineContext context, String nodeId) {
        Flow flow = context.getFlowContext().getFlow();
        AbsNode node = flow.getNodeById(nodeId);
        DataBus dataBus = context.getDataBus();
        dataBus.put(node.getCode(), new HashMap<>());
        List<Edge> outputEdges = flow.getOutputEdges(nodeId);
        for (Edge edge : outputEdges) {
            this.emptyExecuteEdge(context, edge);
        }
    }

    //不满足条件的分支，空跑，主要为了到达等待执行的节点
    private void emptyExecuteEdge(EngineContext context, Edge edge) {
        Flow flow = context.getFlowContext().getFlow();
        List<Edge> allInputEdges = flow.getInputEdges(edge.getTargetNodeId());
        //判断 当前节点 是否为汇聚节点
        if (allInputEdges.size() > 1) {
            this.addArriveEdge(context, edge, allInputEdges, true);
        } else {
            this.emptyExecuteNode(context, edge.getTargetNodeId());
        }
    }

    private void addArriveEdge(EngineContext context, Edge arriveEdge, List<Edge> allInputEdges, boolean isEmptyPipe) {
        Flow flow = context.getFlowContext().getFlow();
        boolean allArrived = false;
        NodeEdgeArriveRecord aggregationListener = null;
        synchronized (context) {
            aggregationListener = context.addArriveEdge(flow.getId(), arriveEdge, isEmptyPipe);
            // 判断是否所有流已到达
            if (aggregationListener.getArrivedEdges().size() == allInputEdges.size()) {
                allArrived = true;
            }
        }
        // 判断是否所有流已到达
        if (allArrived) {
            long emptyPipeCount = aggregationListener.getArrivedEdges().stream().filter(new Predicate<NodeEdgeArriveRecord.EdgeExtension>() {
                @Override
                public boolean test(NodeEdgeArriveRecord.EdgeExtension edgeExtension) {
                    return edgeExtension.isEmptyPipe() == true;
                }
            }).count();
            if (emptyPipeCount == aggregationListener.getArrivedEdges().size()) {
                this.emptyExecuteNode(context, arriveEdge.getTargetNodeId());
            } else {
                //执行当前节点
                this.executeNode(context, arriveEdge.getTargetNodeId());
            }
        }
    }
}